# GF_PY3_CLASS_Wrapper_SQLAlchemy_2_x.py
# Create by GF 2025-04-02 23:42

# 能力补充库 (Ability Supplementary Library)
import sqlite3
# ..................................................
import pandas
import sqlalchemy

# ##################################################

class set_db_path:

    def __init__(self):

        self.db_path = "/data/EXAMPLE.db"  # be Similar to: /data/EXAMPLE.db | C:\EXAMPLE.db

    def update(self, db_path_str:str):
    
        self.db_path = db_path_str
        # ..........................................
        return self  # 返回 self 以实现链式调用

class set_db_host:

    def __init__(self):

        self.db_host = "127.0.0.1"

    def update(self, db_host_str:str):
    
        self.db_host = db_host_str
        # ..........................................
        return self

class set_db_database:

    def __init__(self):

        self.db_database = "working"

    def update(self, db_database_str:str):
    
        self.db_database = db_database_str
        # ..........................................
        return self

class set_db_user:

    def __init__(self):

        self.db_user = "root"

    def update(self, db_user_str:str):
    
        self.db_user = db_user_str
        # ..........................................
        return self

class set_db_password:

    def __init__(self):

        self.db_password = "123456"

    def update(self, db_password_str:str):
    
        self.db_password = db_password_str
        # ..........................................
        return self

class wrapper_sqlalchemy_2_x(object):

    # Example:
    # >>> # Query from SQLite 3
    # >>> w_sqlalchemy = wrapper_sqlalchemy_2_x()
    # >>> w_sqlalchemy.db_path.update("C:\\EXAMPLE.db")
    # >>> result = w_sqlalchemy.sqlite_3_x_query("SELECT * FROM example;")
    # >>> print(result)
    #     id      date   open   high    low  close   volume
    #  0   1  20240731  21.52  21.80  21.40  21.52  1065652
    #  1   2  20240731  21.62  21.92  21.80  21.90  1032323
    # 70  70  20240731  24.30  24.45  24.00  24.20  2050554
    # >>>
    # >>> # Query from MySQL 8
    # >>> w_sqlalchemy = wrapper_sqlalchemy_2_x()
    # >>> w_sqlalchemy.db_host.update("127.0.0.1")
    # >>> w_sqlalchemy.db_database.update("working")
    # >>> w_sqlalchemy.db_user.update("root")
    # >>> w_sqlalchemy.db_password.update("123456")
    # >>> result = w_sqlalchemy.mysql_8_x_query("SELECT * FROM example;")
    # >>> print(result)
    #     id      date   open   high    low  close   volume
    #  0   1  20240731  21.52  21.80  21.40  21.52  1065652
    #  1   2  20240731  21.62  21.92  21.80  21.90  1032323
    # 70  70  20240731  24.30  24.45  24.00  24.20  2050554

    def __init__(self):

        self.db_path     = set_db_path()
        self.db_host     = set_db_host()
        self.db_database = set_db_database()
        self.db_user     = set_db_user()
        self.db_password = set_db_password()

    # ----------------------------------------------
    # Functions for SQLite 3

    def sqlite_3_x_append_xlsx(self, xlsx_path:str, sheet_name:str, db_table_name:str) -> int:

        engine = sqlalchemy.create_engine(f"sqlite:///{self.db_path.db_path}")
        # ..........................................
        Pandas_DataFrame = pandas.read_excel(xlsx_path, sheet_name = sheet_name)
        # ..........................................
        affected_rows = Pandas_DataFrame.to_sql(db_table_name, con = engine, index = False, if_exists = 'append')
        # ..........................................
        return affected_rows

    def sqlite_3_x_append_DataFrame(self, Pandas_DataFrame, db_table_name:str) -> int:

        engine = sqlalchemy.create_engine(f"sqlite:///{self.db_path.db_path}")
        # ..........................................
        affected_rows = Pandas_DataFrame.to_sql(db_table_name, con = engine, index = False, if_exists = 'append')
        # ..........................................
        return affected_rows

    def sqlite_3_x_truncate_table(self, db_table_name:str) -> int:

        # 能力补充函数 (Ability Supplement Function)

        # 连接到 SQLite 3 数据库 (如果文件不存在, 会自动创建)
        connection = sqlite3.connect(self.db_path.db_path)
        # ..........................................
        # 创建一个 Cursor 对象并调用其 execute() 方法执行 SQL 命令
        cursor = connection.cursor()
        # ..........................................
        # Sqlite 3.5x.x 不支持 TRUNCATE TABLE 语句
        cursor.execute(f"DELETE FROM {db_table_name};")
        # ..........................................
        # 提交事务
        connection.commit()
        # ..........................................
        # 关闭 Cursor 和 Connection
        cursor.close()
        connection.close()
        # ..........................................
        return 1

    def sqlite_3_x_query(self, SQL_Statment:str):

        # SQLite 3 in SQLalchemy URI be Similar to: "sqlite:///C:\\EXAMPLE.db"
        engine = sqlalchemy.create_engine(f"sqlite:///{self.db_path.db_path}")
        # ..........................................
        result = pandas.read_sql_query(SQL_Statment, con = engine)
        # ..........................................
        return result

    # ----------------------------------------------
    # Functions for MySQL 8

    def mysql_8_x_append_xlsx(self, xlsx_path:str, sheet_name:str, db_table_name:str) -> int:

        engine = sqlalchemy.create_engine(f"mysql+pymysql://{self.db_user.db_user}:{self.db_password.db_password}@{self.db_host.db_host}:3306/{self.db_database.db_database}")
        # ..........................................
        Pandas_DataFrame = pandas.read_excel(xlsx_path, sheet_name = sheet_name)
        # ..........................................
        affected_rows = Pandas_DataFrame.to_sql(db_table_name, con = engine, index = False, if_exists = 'append', method = 'multi')
        # ..........................................
        return affected_rows

    def mysql_8_x_append_DataFrame(self, Pandas_DataFrame, db_table_name:str) -> int:

        engine = sqlalchemy.create_engine(f"mysql+pymysql://{self.db_user.db_user}:{self.db_password.db_password}@{self.db_host.db_host}:3306/{self.db_database.db_database}")
        # ..........................................
        # pandas.DataFrame.to_sql(name:str, con:object, index = False, if_exists = "append", method:str)
        # 参数说明 (Params):
        # - method = None: 使用标准的 SQL INSERT 语句 (每行一个)。
        # - method = "multi": 在一个 INSERT 语句中传递多个值。
        # 返回值 (Return Value):
        # - int: Number of Affected Rows (method = None / method = "multi").
        # 其它说明 (Other):
        # - If Pandas DataFrame is Empty in (if_exists="append") Mode, return 0.
        affected_rows = Pandas_DataFrame.to_sql(db_table_name, con = engine, index = False, if_exists = 'append', method = 'multi')
        # ..........................................
        return affected_rows

    def mysql_8_x_query(self, SQL_Statment:str):

        # MySQL 8 in SQLalchemy URI be Similar to: "mysql+pymysql://root:abcd1234@127.0.0.1:3306/working"
        engine = sqlalchemy.create_engine(f"mysql+pymysql://{self.db_user.db_user}:{self.db_password.db_password}@{self.db_host.db_host}:3306/{self.db_database.db_database}")
        # ..........................................
        result = pandas.read_sql_query(SQL_Statment, con = engine)
        # ..........................................
        return result

# EOF Signed by GF.
